package MapReduce.Demo12_BookJoin.reduceJoin;

import MapReduce.writableBean.BookBorrowWritable;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import utils.JobSubmit;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @Author:lixinlei
 * @Date:2022/3/24 10:21
 **/
public class BookReduceJoinAllApp {

    public static class BRJAMapper extends Mapper<LongWritable, Text, BookBorrowWritable, NullWritable>{

        BookBorrowWritable outKey = new BookBorrowWritable();
        NullWritable outValue = NullWritable.get();

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

            String line = value.toString();

            String[] fields = line.split("\t");

            if(null!=fields && fields.length>0){
                if(fields.length==2){
                    //如果字段数为2 且第一个字段的长度为2 此时为person数据
                    if(fields[0].length()==2){

                        outKey.setPersonId(fields[0]);
                        outKey.setPersonName(fields[1]);
                        outKey.setFlag("person");

                    }else if(fields[0].length()==4){
                        //如果字段数为2 且第一个字段的长度为4 此时为book数据
                        outKey.setBookId(fields[0]);
                        outKey.setBookName(fields[1]);
                        outKey.setFlag("book");
                    }
                }else if(fields.length==3){
                    //如果字段数为3，此时为借阅记录数据
                    outKey.setBookId(fields[0]);
                    outKey.setPersonId(fields[1]);
                    outKey.setDateTime(fields[2]);
                    outKey.setFlag("borrow");
                }
            }

            context.write(outKey,outValue);

        }
    }

    public static class BRJAReducer extends Reducer<BookBorrowWritable,NullWritable,BookBorrowWritable,NullWritable>{

        //初始化所有人员的借阅记录(包含了人员姓名)的列表
        List<BookBorrowWritable> borrowAndNameList = new ArrayList<BookBorrowWritable>();

        //初始化所有的图书信息
        Map<String,String> bookMap = new HashMap<String,String>();

        //初始化所有的图书信息
        Map<String,String> personMap = new HashMap<String,String>();

        @Override
        protected void reduce(BookBorrowWritable key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {

            //初始化同一个人员下的人员姓名
            String personName = "";
            //初始化同一个人员下的所有借阅记录数据的列表
            List<BookBorrowWritable> borrowList = new ArrayList<BookBorrowWritable>();

            /**
             *  数据样例
             *      1001    01  2022-03-24
             *      01  tom
             *      1002    01  2022-03-23
             *  或者
             *      1001    活着
             *      1002    高血压的预防
             *      1003    颈椎病的注意事项
             *
             */

            for (NullWritable value : values) {
                String flag = key.getFlag();
                if(flag.equals("book")){
                    //把所有的图书信息，封装到全局的map中
                    bookMap.put(key.getBookId(),key.getBookName());
                }else{
                    //如果是借阅记录，先封装成一个集合，后面再遍历
                    if(flag.equals("borrow")){

                        try {
                            BookBorrowWritable borrow = new BookBorrowWritable();
                            BeanUtils.copyProperties(borrow, key);
//                            borrowList.add(borrow);

                            borrowAndNameList.add(borrow);
                        } catch (Exception e) {
                            e.printStackTrace();
                        }

                    }else if(flag.equals("person")){
//                        personName = key.getPersonName();
                        personMap.put(key.getPersonId(),key.getPersonName());
                    }
                }
            }

//            for (BookBorrowWritable borrowInfo : borrowList) {
//                borrowInfo.setPersonName(personName);
//                borrowAndNameList.add(borrowInfo);
//            }

        }

        @Override
        protected void cleanup(Context context) throws IOException, InterruptedException {
            for (BookBorrowWritable borrowAndName : borrowAndNameList) {
                borrowAndName.setBookName(bookMap.get(borrowAndName.getBookId()));
                borrowAndName.setPersonName(personMap.get(borrowAndName.getPersonId()));
                context.write(borrowAndName,NullWritable.get());
            }
        }
    }

    public static void main(String[] args) {

//        JobSubmit.submitBaseJob(
//                BookReduceJoinAllApp.class,
//                args,
//                BookReduceJoinSecondGroupComparator.class
//        );

        JobSubmit.submitBaseJob(BookReduceJoinAllApp.class,args);
    }

}
